#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include "shmem.h"
#include "cJSON.h"
#include <list.h>
#include "msg_queue_peer.h"
#include <pthread.h>

#define ADDRESS "tcp://localhost:1883"
#define PUBCLIENTID "ExampleCClientPub"
#define SUBCLIENTID "ExampleDClientSub"
#define PUBTOPIC "up"
#define SUBTOPIC "ctl_fx2"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L

pthread_t tid_sub;
pthread_t tid_pub;
char *submsg[128];
char *pubmsg[128];

struct student
{
    char name[32];
    int num;
    float avegrade;
};

struct msg
{
    BOOL flag;
    int num;
    struct student stu[10];
};

struct coremsg 
{
    struct msg msg;
    struct list_head list;
};

struct ctl
{
    int key;
    char name[32];
    int val;
};

struct msg_ctl
{
    long type;
    struct ctl ctl;
};

struct msg_ctl msg_ctl = {0,{}};

volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;
    char buf[1024];
    struct msg_ctl msg_ctl;
    msg_ctl.type = 1;
    
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf(" message: ");

    payloadptr = message->payload;
    for (i = 0; i < message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');
    strcpy(buf,message->payload);
    
    buf[strlen(buf)] = '\0';

    cJSON *root = cJSON_Parse(buf);

    cJSON *key = cJSON_GetObjectItem(root, "key");
    msg_ctl.ctl.key = key->valueint;

    cJSON *name = cJSON_GetObjectItem(root, "name");
    strcpy(msg_ctl.ctl.name,name->valuestring);

    cJSON *val = cJSON_GetObjectItem(root, "val");
    msg_ctl.ctl.val = val->valueint;
    printf("send\n");
    printf("%d %s %d\n",msg_ctl.ctl.key,msg_ctl.ctl.name,msg_ctl.ctl.val);
    msg_queue_send("b.c", &msg_ctl, sizeof(msg_ctl), 0);

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

void *subcribe(void *arg) //订阅消息
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;

    MQTTClient_create(&client, ADDRESS, SUBCLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           SUBTOPIC, SUBCLIENTID, QOS);
    MQTTClient_subscribe(client, SUBTOPIC, QOS);

    do
    {
        ch = getchar();
    } while (ch != 'Q' && ch != 'q');

    MQTTClient_unsubscribe(client, SUBTOPIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return (void *)rc;
}

void *publish(void *arg)
{
    printf("publish\n");
    struct shm_param para;
    char msg[2048];
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    MQTTClient_create(&client, ADDRESS, PUBCLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    //定义头结点
    struct list_head head;
    //初始化内核链表
    INIT_LIST_HEAD(&head);

    shm_init(&para, "a.c", 1024);
    struct msg *p = shm_getaddr(&para);
    BOOL flag = p->flag;
    int i = 0;
    while (1)
    {    
        while (flag == p->flag)
        {
            sleep(2);
            printf("wait to publish\n");  
        };
        flag = p->flag;

        cJSON* root = cJSON_CreateObject();
        cJSON* data = cJSON_CreateArray();

        for (int i = 0; i < p->num; i++)
        {
            cJSON* item = cJSON_CreateObject();
            cJSON_AddStringToObject(item, "name", p->stu[i].name);
            cJSON_AddNumberToObject(item, "num", p->stu[i].num);
            cJSON_AddNumberToObject(item, "avegrade", p->stu[i].avegrade);
            cJSON_AddItemToArray(data, item);
        }

        cJSON_AddItemToObject(root, "data", data);

        strcpy(msg,cJSON_Print(root));

        printf("-----------------------------\n");
        printf("corelist:\n");

        struct coremsg* coremsg = (struct coremsg*)malloc(sizeof(struct coremsg));
        coremsg->msg = *p;
        list_add(&coremsg->list,&head);
        
        struct coremsg *tmp;
        struct list_head *pos;
        int i = 0;

        list_for_each(pos,&head)
        {
            printf("%d\n",i++);
            tmp = list_entry(pos,struct coremsg,list);//返回数据段的首地址
            printf("%d\n",tmp->msg.num);
            for (int i = 0; i < tmp->msg.num; i++)
            {   
                 printf("name = %s, num = %d, avegrade = %f\n", tmp->msg.stu[i].name, tmp->msg.stu[i].num, tmp->msg.stu[i].avegrade);
            }
        }
        printf("-----------------------------\n");
    
        pubmsg.payload = msg;
        pubmsg.payloadlen = (int)strlen(msg);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;

        MQTTClient_publishMessage(client, PUBTOPIC, &pubmsg, &token);

        sleep(1);
        cJSON_Delete(root);
    }

    // MQTTClient_unsubscribe(client, TOPIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    free(msg);
    
    return (void*)rc;
}

int main(int argc, char *argv[])
{
    pthread_create(&tid_pub, NULL, subcribe, NULL);
    pthread_create(&tid_sub, NULL, publish, NULL);
    pthread_detach(tid_pub);
    pthread_detach(tid_sub);
    while (1)
    {
        pause();
    }
        return 0;
}
